package org.porkbelly.json;

import java.io.IOException;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;

import org.apache.pig.EvalFunc;
import org.apache.pig.data.NonSpillableDataBag;
import org.apache.pig.data.Tuple;

/**
 * <p>
 * JsonMapToTuple will convert JSON data loaded with Twitter's Elephant-bird
 * and convert it to Map. First the data is loaded with JsonLoader like
 * this:</p>
 * <p>mydata = load '/inputdata.json' USING com.twitter.elephantbird.pig.load.JsonLoader('-nestedLoad')
 *  as (json:map[]);</p>
 * <p>Basically, this lets you take a list of stuff and pivot it into rows
 * of Map data. After that you can filter or group the data as you need.</p>
 * 
 * @author peter lin
 *
 */
@SuppressWarnings("rawtypes")
public class JsonBagToMap extends EvalFunc<Map> {

	public JsonBagToMap() {
		
	}
	
	@Override
	public Map exec(Tuple input) throws IOException {
		if (input == null || input.size() <= 0) {
			return null;
		}
		NonSpillableDataBag bag = null;
		try {
			Map<String, Object> output = new LinkedHashMap<String, Object>();

			Object v = input.get(0);
			bag = (NonSpillableDataBag)v;
			Iterator bagIterator = bag.iterator();
			while (bagIterator.hasNext()) {
				Object t = bagIterator.next();
				if (t instanceof Tuple) {
					Tuple tuple = (Tuple)t;
					for (int i=0; i < tuple.size(); i++) {
						output.put((String)tuple.get(i), new Integer(0));
					}
				}
			}

			return output;
		} catch (ClassCastException e) {
			throw new RuntimeException("Input type must be a Bag: " + input.get(0).getClass().getName() +
					" \n tuple size: " + input.size() +
					" \n bag size: " + bag.size());
		} catch (ArrayIndexOutOfBoundsException e) {
			throw new RuntimeException(
					"Function input must have even number of parameters");
		} catch (Exception e) {
			throw new RuntimeException("Error while creating a map", e);
		}
	}
}
